package cn.doitedu.api;


import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.text.DecimalFormat;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/2/22
 * @Desc: 学大数据，上多易教育
 *
 *    socket输入数据:
 *       a,18
 *       a,20
 *       b,30
 *       c,30
 *       a,40
 *       b,25
 *
 *    利用map算子，来实现如下计算需求：
 *        按照相同 name，对输入的数据去不断更新计算  score的平均值
 *        要求用map算子来实现
 *
 *
 **/
public class _11_State_First_Exercise {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<String> stream = env.socketTextStream("doitedu", 9999);

        // 首先,按照name 来分组
        stream.keyBy(new KeySelector<String, String>() {
            @Override
            public String getKey(String value) throws Exception {
                return value.split(",")[0];
            }
        })
        .map(new RichMapFunction<String, String>() {   // 核心计算逻辑

            ValueState<Integer> cntState;
            ValueState<Integer> sumState;

            DecimalFormat decimalFormat;

            @Override
            public void open(Configuration parameters) throws Exception {
                // 用来记录数据条数
                cntState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("cnt", Integer.class));
                // 用来记录分数总和
                sumState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("sum", Integer.class));

                //
                this.decimalFormat = new DecimalFormat("#.##");


            }

            @Override
            public String map(String value) throws Exception {
                // b,25

                // 切出name
                String name = value.split(",")[0];

                // 先切出score
                int score = Integer.parseInt(value.split(",")[1]);

                // 取出这个人此前的分数累加值
                int sum = sumState.value() == null ? 0 : sumState.value();
                // 累加当前的score
                sum = sum + score;

                // 将最新的分数之和，要更新到状态中
                sumState.update(sum);

                // 对这个人的数据条数递增
                int cnt = cntState.value() == null ?  1 : cntState.value() + 1;
                cntState.update(cnt);


                return name + " : " + decimalFormat.format((double)sum/cnt);
            }
        }).print();


        env.execute();

    }
}
